/*
 * Copyright (2023) The Delta Lake Project Authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.delta.kernel.internal.actions;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.function.Function;

import io.delta.kernel.data.Row;
import io.delta.kernel.types.IntegerType;
import io.delta.kernel.types.LongType;
import io.delta.kernel.types.StringType;
import io.delta.kernel.types.StructType;
import static io.delta.kernel.utils.Utils.requireNonNull;

import io.delta.kernel.internal.data.PojoRow;
import io.delta.kernel.internal.deletionvectors.Base85Codec;
import io.delta.kernel.internal.fs.Path;
import static io.delta.kernel.internal.util.InternalUtils.checkArgument;

/**
 * Information about a deletion vector attached to a file action.
 */
public class DeletionVectorDescriptor {

    ////////////////////////////////////////////////////////////////////////////////
    // Static Fields / Methods
    ////////////////////////////////////////////////////////////////////////////////
    public static DeletionVectorDescriptor fromRow(Row row) {
        if (row == null) return null;

        final String storageType = requireNonNull(row, 0, "storageType").getString(0);
        final String pathOrInlineDv = requireNonNull(row, 1, "pathOrInlineDv").getString(1);
        final Optional<Integer> offset = Optional.ofNullable(
            row.isNullAt(2) ? null : row.getInt(2));
        final int sizeInBytes = requireNonNull(row, 3, "sizeInBytes").getInt(3);
        final long cardinality = requireNonNull(row, 4, "cardinality").getLong(4);

        return new DeletionVectorDescriptor(storageType, pathOrInlineDv, offset,
            sizeInBytes, cardinality);
    }

    // Markers to separate different kinds of DV storage.
    public static final String PATH_DV_MARKER = "p";
    public static final String INLINE_DV_MARKER = "i";
    public static final String UUID_DV_MARKER = "u";

    /**
     * String that is used in all file names generated by deletion vector store
     */
    static final String DELETION_VECTOR_FILE_NAME_CORE = "deletion_vector";

    public static final StructType READ_SCHEMA = new StructType()
        .add("storageType", StringType.INSTANCE, false /* nullable*/)
        .add("pathOrInlineDv", StringType.INSTANCE, false /* nullable*/)
        .add("offset", IntegerType.INSTANCE, true /* nullable*/)
        .add("sizeInBytes", IntegerType.INSTANCE, false /* nullable*/)
        .add("cardinality", LongType.INSTANCE, false /* nullable*/);

    ////////////////////////////////////////////////////////////////////////////////
    // Instance Fields / Methods
    ////////////////////////////////////////////////////////////////////////////////

    /**
     * Indicates how the DV is stored.
     * Should be a single letter (see [[pathOrInlineDv]] below.)
     */
    private final String storageType;

    /**
     * Contains the actual data that allows accessing the DV.
     * <p>
     * Three options are currently supported:
     * - `storageType="u"` format: `<random prefix - optional><base85 encoded uuid>`
     * The deletion vector is stored in a file with a path relative to
     * the data directory of this Delta Table, and the file name can be
     * reconstructed from the UUID.
     * The encoded UUID is always exactly 20 characters, so the random
     * prefix length can be determined any characters exceeding 20.
     * - `storageType="i"` format: `<base85 encoded bytes>`
     * The deletion vector is stored inline in the log.
     * - `storageType="p"` format: `<absolute path>`
     * The DV is stored in a file with an absolute path given by this
     * url.
     */
    private final String pathOrInlineDv;

    /**
     * Start of the data for this DV in number of bytes from the beginning of the file it is stored
     * in.
     * <p>
     * Always None when storageType = "i".
     */
    private final Optional<Integer> offset;

    /**
     * Size of the serialized DV in bytes (raw data size, i.e. before base85 encoding).
     */
    private final int sizeInBytes;

    /**
     * Number of rows the DV logically removes from the file.
     */
    private final long cardinality;

    public DeletionVectorDescriptor(
        String storageType,
        String pathOrInlineDv,
        Optional<Integer> offset,
        int sizeInBytes,
        long cardinality) {
        this.storageType = storageType;
        this.pathOrInlineDv = pathOrInlineDv;
        this.offset = offset;
        this.sizeInBytes = sizeInBytes;
        this.cardinality = cardinality;
    }

    public String getStorageType() {
        return storageType;
    }

    public String getPathOrInlineDv() {
        return pathOrInlineDv;
    }

    public Optional<Integer> getOffset() {
        return offset;
    }

    public int getSizeInBytes() {
        return sizeInBytes;
    }

    public long getCardinality() {
        return cardinality;
    }

    public String getUniqueId() {
        String uniqueFileId = storageType + pathOrInlineDv;
        if (offset.isPresent()) {
            return uniqueFileId + "@" + offset;
        } else {
            return uniqueFileId;
        }
    }

    public boolean isInline() {
        return storageType == INLINE_DV_MARKER;
    }

    public boolean isOnDisk() {
        return !isInline();
    }

    public byte[] inlineData() {
        checkArgument(isInline(), "Can't get data for an on-disk DV from the log.");
        // The sizeInBytes is used to remove any padding that might have been added during encoding.
        return Base85Codec.decodeBytes(pathOrInlineDv, sizeInBytes);
    }

    public String getAbsolutePath(String tableLocation) {
        checkArgument(isOnDisk(), "Can't get a path for an inline deletion vector");
        if (storageType.equals(UUID_DV_MARKER)) {
            // If the file was written with a random prefix, we have to extract that,
            // before decoding the UUID.
            int randomPrefixLength = pathOrInlineDv.length() - Base85Codec.ENCODED_UUID_LENGTH;
            String randomPrefix = pathOrInlineDv.substring(0, randomPrefixLength);
            String encodedUuid = pathOrInlineDv.substring(randomPrefixLength);
            UUID uuid = Base85Codec.decodeUUID(encodedUuid);
            return assembleDeletionVectorPath(tableLocation, uuid, randomPrefix).toString();
        } else if (storageType.equals(PATH_DV_MARKER)) {
            // Since there is no need for legacy support for relative paths for DVs,
            // relative DVs should *always* use the UUID variant.
            try {
                URI parsedUri = new URI(pathOrInlineDv);
                checkArgument(parsedUri.isAbsolute(),
                    "Relative URIs are not supported for DVs");
                return new Path(parsedUri).toString();
            } catch (URISyntaxException e) {
                throw new RuntimeException("Couldn't parse uri:\n" + e);
            }
        } else {
            throw new RuntimeException("A uri " + pathOrInlineDv +
                " which cannot be turned into a relative path as found in the transaction log");
        }
    }

    /**
     * Return the unique path under `parentPath` that is based on `id`.
     * <p>
     * Optionally, prepend a `prefix` to the name.
     */
    private Path assembleDeletionVectorPath(
        String targetParentPath,
        UUID id,
        String prefix) {
        String fileName = String.format("%s_%s.bin", DELETION_VECTOR_FILE_NAME_CORE, id.toString());
        if (prefix.length() > 0) {
            return new Path(new Path(targetParentPath, prefix), fileName);
        } else {
            return new Path(targetParentPath, fileName);
        }
    }

    @Override
    public String toString() {
        return String.format(
            "DeletionVectorDescriptor(storageType=%s, pathOrInlineDv=%s, offset=%s, " +
                "sizeInBytes=%s, cardinality=%s)",
            storageType, pathOrInlineDv, offset, sizeInBytes, cardinality);
    }

    @Override
    public boolean equals(Object o) {
        if (o == this) return true;
        if (!(o instanceof DeletionVectorDescriptor)) return false;
        DeletionVectorDescriptor dv = (DeletionVectorDescriptor) o;
        return Objects.equals(storageType, dv.storageType) &&
            Objects.equals(pathOrInlineDv, dv.pathOrInlineDv) &&
            Objects.equals(offset, dv.offset) &&
            this.sizeInBytes == dv.sizeInBytes &&
            this.cardinality == dv.cardinality;
    }

    @Override
    public int hashCode() {
        return Objects.hash(storageType, pathOrInlineDv, offset, sizeInBytes, cardinality);
    }

    public Row toRow() {
        return new PojoRow<>(
            this,
            READ_SCHEMA,
            ordinalToAccessor);
    }

    ////////////////////////////////////////////////////////////////////////////////
    // Static fields to create PojoRows
    ////////////////////////////////////////////////////////////////////////////////

    private static final Map<Integer, Function<DeletionVectorDescriptor, Object>>
        ordinalToAccessor = new HashMap<>();

    static {
        ordinalToAccessor.put(0, (a) -> a.getStorageType());
        ordinalToAccessor.put(1, (a) -> a.getPathOrInlineDv());
        ordinalToAccessor.put(2, (a) -> a.getOffset().orElse(null));
        ordinalToAccessor.put(3, (a) -> a.getSizeInBytes());
        ordinalToAccessor.put(4, (a) -> a.getCardinality());
    }
}
